package com.hucais.etl.common.dao

import com.hucais.core.utils.DefaultPropertiesUtil
import com.hucais.etl.job.{OdsBookComment, OdsBookInformation, OdsBookStore}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.{Dataset, SparkSession}
import org.elasticsearch.spark.sql.EsSparkSQL

object OdsEsDao {
  private val queryAll =
    """
      |{
      |	"query": {
      |		"bool": {
      |			"must": {"term": {"data_source": "%s"}}
      |		}
      |	},
      |	"_source": {
      |		"includes": [%s]
      |	}
      |}
      |""".stripMargin

  private val queryWithTime =
    """
      |{
      |	"query": {
      |		"bool": {
      |			"must": {"term": {"data_source": "%s"}},
      |			"filter": {
      |			  "range": {
      |					"acquisition_time": {
      |						"gte": "%s 00:00:00",
      |						"lte": "%s 23:59:59",
      |						"format":"yyyy-MM-dd HH:mm:ss",
      |				    "time_zone":"+08:00"
      |					}
      |				}
      |			}
      |		}
      |	},
      |	"_source": {
      |		"includes": [%s]
      |	}
      |}
      |""".stripMargin

  /**
   * 获取ODS-BookInformation数据
   *
   * @param sparkSession esSpark API需要的sparksession
   * @param dataSource   数据来源(或全文检索的一级渠道)
   * @param colList      查询index的字段
   * @param day          查询哪一天的数据(格式:yyyy-MM-dd)
   * @return
   */
  def getBook(sparkSession: SparkSession, dataSource: String, colList: String, day: String): Dataset[OdsBookInformation] = {
    import sparkSession.implicits._
    val query = if (StringUtils.isNotBlank(day)) queryWithTime.format(dataSource, day, day, colList) else queryAll.format(dataSource, colList)
    EsSparkSQL.esDF(sparkSession, DefaultPropertiesUtil.get("ods.book.information"), query).as[OdsBookInformation]
  }

  /**
   * 获取ODS-BookSotre数据
   *
   * @param sparkSession esSpark API需要的sparksession
   * @param dataSource   数据来源(或全文检索的一级渠道)
   * @param colList      查询index的字段
   * @param day          查询哪一天的数据(格式:yyyy-MM-dd)
   * @return
   */
  def getStore(sparkSession: SparkSession, dataSource: String, colList: String, day: String): Dataset[OdsBookStore] = {
    import sparkSession.implicits._
    val query = if (StringUtils.isNotBlank(day)) queryWithTime.format(dataSource, day, day, colList) else queryAll.format(dataSource, colList)
    EsSparkSQL.esDF(sparkSession, DefaultPropertiesUtil.get("ods.book.store"), query).as[OdsBookStore]
  }

  /**
   * 获取ODS-BookComment数据
   *
   * @param sparkSession esSpark API需要的sparksession
   * @param dataSource   数据来源(或全文检索的一级渠道)
   * @param colList      查询index的字段
   * @param day          查询哪一天的数据(格式:yyyy-MM-dd)
   * @return
   */
  def getComment(sparkSession: SparkSession, dataSource: String, colList: String, day: String): Dataset[OdsBookComment] = {
    import sparkSession.implicits._
    val query = if (StringUtils.isNotBlank(day)) queryWithTime.format(dataSource, day, day, colList) else queryAll.format(dataSource, colList)
    EsSparkSQL.esDF(sparkSession, DefaultPropertiesUtil.get("ods.book.comment"), query).as[OdsBookComment]
  }

}
